package com.test.flink_transforms;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class F06ReduceFunction {

    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port", 8888);
        StreamExecutionEnvironment webUI = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);

        DataStreamSource<String> data = webUI.socketTextStream("127.0.0.1", 8899);

        DataStream<Tuple2<String, Integer>> res = data.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String s, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] words = s.split("\\s+");

                for (String word : words) {
                    out.collect(Tuple2.of(word, 1));
                }
            }
        });

        KeyedStream<Tuple2<String, Integer>, String> keyed = res.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            @Override
            public String getKey(Tuple2<String, Integer> value) throws Exception {
                return value.f0;
            }
        });

//        SingleOutputStreamOperator<String> ds = key.sum(1);

        //数据聚合前后保持一致
        SingleOutputStreamOperator<Tuple2<String, Integer>> ds = keyed.reduce(new ReduceFunction<Tuple2<String, Integer>>() {

            int count = 1;
            double money = 0d;

            /**
             *
             * @param value1 中间数据
             * @param value2 后面的值
             * @return
             * @throws Exception
             */
            @Override
            public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception {
                // 求和 sum
//                value1.f1 = value1.f1 + value2.f1;
                // 最大值 max
//                value1.f1 = value1.f1 > value2.f1 ? value1.f1 : value2.f1;
                // 最大By maxBy
//                if(value1.f1 > value2.f1){
//                    return value1;
//                } else {
//                    return value2;
//                }

                // 自定义 求平均
                count += 1;
                money = value1.f1;

                double total = money * (count - 1) + value2.f1;
                double avg = total / count;

                return Tuple2.of(value1.f0, (int)avg);
            }
        });

        ds.print();

        webUI.execute();
    }
}
